package com.mattgarner.jaddas.node.index;

import com.mattgarner.jaddas.common.ColumnType;
import com.mattgarner.jaddas.common.ResultSet;
import com.mattgarner.jaddas.common.ResultSetException;
import com.mattgarner.jaddas.common.ResultSetHeader;
import com.mattgarner.jaddas.dataset.DataSetPartition;
import com.mattgarner.jaddas.node.LocalDataProvider;
import com.mattgarner.jaddas.node.data.LocalTable;
import com.mattgarner.jaddas.node.data.TableException;
import com.mattgarner.jaddas.node.data.TableHeader;
import com.sleepycat.je.Cursor;
import com.sleepycat.je.Database;
import com.sleepycat.je.DatabaseEntry;
import com.sleepycat.je.LockMode;
import com.sleepycat.je.OperationStatus;

public class LocalIndex {

	 
	private LocalTable table;
	private TableHeader header;
	
	private DataSetPartition dsPartition;
	private String dataStore;
	private String dataName;
	
	private LocalDataProvider dataProvider;
	private Database dbLocalIndexData;
	
	public LocalIndex (DataSetPartition dsPartition, String dataStore, String dataName) throws IndexException {
		this.dsPartition = dsPartition;
		this.dataStore = dataStore;
		this.dataName = dataName;
		
		if(dsPartition.getLocalTable() == null) {
			throw new IndexException("LocalIndex: localTable must be specified.");
		}
		table = dsPartition.getLocalTable();
		header = table.getHeader();
		
		dataProvider = LocalDataProvider.getInstance();
		dbLocalIndexData = dataProvider.openDatabase(dataStore, dataName, true);
		
	}
	
	public final ColumnIndexStatistics[] generateColumnStatistics() throws IndexException {
		try {
			// data structure for holding column index
			//HashMap<Integer, HashMap<String, Boolean>> columnStringIndex = new HashMap<Integer, HashMap<String, Boolean>>();
			
			Cursor cursor = table.getNewDatabaseCursor();
			DatabaseEntry key = new DatabaseEntry();
			DatabaseEntry data = new DatabaseEntry();

			ColumnIndexStatistics[] colIndexStats = new ColumnIndexStatistics[header.getColumnCount()];
			
			int rowCounter = 0;
			while(cursor.getNext(key, data, LockMode.DEFAULT) == OperationStatus.SUCCESS) {
				
				byte[] rowBytes = data.getData();
				this.updateColumnStatisticsFromRowBytes(colIndexStats, rowBytes);
				
				rowCounter++;
			}
			
			cursor.close();
			
			
			// update cardinality stats
			/*
			Integer[] stringColumns = columnStringIndex.keySet().toArray(new Integer[0]);
			for(int a=0; a<stringColumns.length; a++) {
				int col = stringColumns[a];
				
				int cardinality = columnStringIndex.get(col).keySet().size();
				colIndexStats[col].setCardinality(cardinality);
				
				// free some mem (this may not be efficient)
				columnStringIndex.remove(col);
			}
			*/
			
			return colIndexStats;
		} catch (Exception e) {
			////
		}
		
		return null;
	}
	
	private final void updateColumnStatisticsFromRowBytes(ColumnIndexStatistics[] colIndexStats, byte[] rowBytes) throws IndexException {

		try {
			int columnCount = header.getColumnCount();
			for(int col=0;col<columnCount;col++) {
				
				Object value = null;
				
				
				if(header.getColumnBasicType(col) == ColumnType.COLUMN_TYPE_INT) {
					value = new Integer(table.getIntValue(rowBytes, col));
				} else if(header.getColumnBasicType(col) == ColumnType.COLUMN_TYPE_SHORT) {
					value = new Integer(table.getShortValue(rowBytes, col));
				} else if(header.getColumnBasicType(col) == ColumnType.COLUMN_TYPE_VARCHAR) {
					value = table.getStringValue(rowBytes, col);
				} 

				updateColumnStatisticsByValue(colIndexStats[col], col, value);
			
			}
		} catch (Exception e) {
			
		}

	}
	
	public void updateColumnStatisticsByValue(int col, Object value) throws IndexException {
		this.updateColumnStatisticsByValue(dsPartition.getColumnIndexStatistic(col), col, value);
	}
	
	private void updateColumnStatisticsByValue(ColumnIndexStatistics colStats, int col, Object value) throws IndexException {
		try {
			Integer intValue = null;
			String stringValue = null;	
			
			if(value instanceof Integer) {
				intValue = (Integer)value;
			} else if(value instanceof String) {
				stringValue = (String)value;
			}
			
			if(colStats == null) { 
				colStats = new ColumnIndexStatistics();
				colStats.setColumnID(col);
				colStats.setColumnType(header.getColumnBasicType(col));
			}
			
			if((intValue == null) && (stringValue == null)) {
				return;
			}
			
			
			// find max and min values for numerics
			if(intValue != null) {
				if(colStats.getMaxValue() == null) {
					colStats.setMaxValue(intValue);
				}
				if(intValue > colStats.getMaxValue()) {
					colStats.setMaxValue(intValue);
				}
				if(colStats.getMinValue() == null) {
					colStats.setMinValue(intValue);
				}
				if(intValue < colStats.getMinValue()) {
					colStats.setMinValue(intValue);
				}
				
				// also collect stats on numeric columns as if they were strings
				stringValue = intValue.toString(); 
			}
			
			// get statistics for strings
			if(stringValue != null) {
				
				// use this hashmap to determine cardinality
				/*
				HashMap<String, Boolean> columnStringValues = columnStringIndex.get(col);
				if(columnStringValues == null) {
					columnStringValues = new HashMap<String, Boolean>();
				}
				
				columnStringValues.put(stringValue, true);
				columnStringIndex.put(col, columnStringValues);
				*/
				
				// use character mask to determine *approximate* cardinality
				boolean maskMatch = colStats.checkStringAgainstCharacterMask(stringValue);
				if(!maskMatch) {
					int currentCardinality = colStats.getCardinality();
					colStats.setCardinality(currentCardinality + 1);
				}
				
				// update maximum length
				if(stringValue.length() > colStats.getLongestStringLength()) {
					colStats.setLongestStringLength(stringValue.length());
				}
				
				// update string character mask
				colStats.addStringToCharacterMask(stringValue);
			}
		} catch (Exception e) {
			throw new IndexException("updateColumnStatisticsByValue: " + e.getMessage());
		}

	}
	
	public void buildBitmapIndex(String[] columns) throws IndexException {
		
		// check that specified columns exist and get the column ID
		int[] columnIDs = new int[columns.length];
		for(int a=0;a<columns.length;a++) {
			int colID = header.getColumnIndexByName(columns[a]);
			if(colID == -1) {
				throw new IndexException("buildBitmapIndex: '" + columns[a] + "' is not a valid column name.");
			} else {
				columnIDs[a] = colID;
			}
		}
		
		try {
			int columnCount = header.getColumnCount();
			int rowCount = (int)table.getFullRowCount();
			
			BitmapIndex[] bitmapIndices = new BitmapIndex[columnCount];
			
			Cursor cursor = table.getNewDatabaseCursor();
			DatabaseEntry key = new DatabaseEntry();
			DatabaseEntry data = new DatabaseEntry();

			// fetch each row of the table			
			int rowCounter = 0;
			while(cursor.getNext(key, data, LockMode.DEFAULT) == OperationStatus.SUCCESS) {
				
				byte[] rowBytes = data.getData();
				
				for(int b=0;b<columnIDs.length;b++) {
					int col = columnIDs[b];
					
					
					if(bitmapIndices[col] == null) {
						bitmapIndices[col] = new BitmapIndex(dataProvider, dataStore, dataName, columns[b], rowCount);
						bitmapIndices[col].clearIndex();
					}
					
					String stringValue = null;
					
					if(header.getColumnBasicType(col) == ColumnType.COLUMN_TYPE_INT) {
						stringValue = new Integer(table.getIntValue(rowBytes, col)).toString();
					} else if(header.getColumnBasicType(col) == ColumnType.COLUMN_TYPE_SHORT) {
						stringValue = new Integer(table.getShortValue(rowBytes, col)).toString();
					} else if(header.getColumnBasicType(col) == ColumnType.COLUMN_TYPE_VARCHAR) {
						stringValue = table.getStringValue(rowBytes, col);
					} 
				
					bitmapIndices[col].addValueToIndex(stringValue, rowCounter);
					
				}
				
				rowCounter++;
			}
			cursor.close();
			  
		
		} catch (Exception e) {
			throw new IndexException("buildBitmapIndex: " + e.getMessage());
		}
		
	}
	
	public final void addValueToBitmapIndex(int col, int position, Object value) throws IndexException {
		try {
			BitmapIndex bitmapIndex = new BitmapIndex(dataProvider, dataStore, dataName, header.getColumnName(col), (int)table.getFullRowCount());
			if(!(value instanceof String)) {
				value = value.toString();
			}
			bitmapIndex.addValueToIndex((String)value, position);
		} catch (Exception e) {
			throw new IndexException("addValueToBitmapIndex: " + e.getMessage());
		}
	}
	
	public ResultSet convertColumnStatsToResultSet(ColumnIndexStatistics[] colIndexStats, String stringEncoding) throws ResultSetException {
		// create ResultSet
		try { 
			ResultSetHeader rsHeader = new ResultSetHeader(stringEncoding);
			rsHeader.addIntColumn("ID");
			rsHeader.addStringColumn("Type");
			rsHeader.addStringColumn("Column Name");
			rsHeader.addStringColumn("Minimum");
			rsHeader.addStringColumn("Maximum");
			rsHeader.addIntColumn("Cardinality");
			rsHeader.addIntColumn("Longest Value");
			
			ResultSet resultSet = new ResultSet(rsHeader);
			
						
			for(int a=0; a<colIndexStats.length;a++) {
				int headerColumnID = colIndexStats[a].getColumnID();
				resultSet.setValue(a, 0, headerColumnID);
				resultSet.setValue(a, 1, header.getColumnTypeString(headerColumnID));
				resultSet.setValue(a, 2, header.getColumnName(headerColumnID));
				resultSet.setValue(a, 3, colIndexStats[a].getMinValueString());
				resultSet.setValue(a, 4, colIndexStats[a].getMaxValueString());
				resultSet.setValue(a, 5, colIndexStats[a].getCardinality());
				resultSet.setValue(a, 6, colIndexStats[a].getLongestStringLength());
			}
			
			return resultSet;
		} catch (Exception e) {
			throw new ResultSetException("ResultSet Exception: " + e.getMessage());
		}	
	}
	
	
	
}
